package pulsar

import (
	"context"
	"errors"
	"runtime/debug"
	"sync"
	"time"

	"github.com/apache/pulsar-client-go/pulsar"
	"github.com/apache/pulsar-client-go/pulsar/log"

	"knowledge_admin/pkg/adapters/queue/pubsub"
)

const (
	errSleepTime         = 2 * time.Second
	ACK_TYPE_NOT_PULLACK = 0 //拉取msg时候不ACK，使用方自主ACK
	ACK_TYPE_PULLACK     = 1 //拉取消息就ACK，使用方可不用ACK
)

// ErrMsgIgnore ignore pt_msg error
var ErrMsgIgnore = errors.New("do not consume pt_msg")

type sub struct {
	client        pulsar.Client
	consumer      pulsar.Consumer
	notPtMsg      bool
	ackType       int
	subscribeName string
	topic         string
	spanName      string
	wg            sync.WaitGroup
	once          sync.Once
	cancelFunc    context.CancelFunc
}

func (s *sub) Close() error {
	s.once.Do(func() {
		if s.cancelFunc != nil {
			s.cancelFunc()
		}
		s.wg.Wait()

		s.consumer.Close()
		s.client.Close()
	})

	return nil
}

func (s *sub) Receive(ctx context.Context) (pubsub.Message, error) {
	for {
		var pmsg pulsar.Message
		var err error

		pmsg, err = s.consumer.Receive(ctx)

		if err != nil {
			return nil, err
		}

		msg := &message{
			Message:       pmsg,
			ctx:           ctx,
			ack:           s.consumer.Ack,
			subscribeName: s.subscribeName,
		}

		return msg, nil
	}
}

func (s *sub) Start(ctx context.Context, handler pubsub.MessageHandler) (err error) {
	s.wg.Add(1)

	defer func() {
		s.wg.Done()
		s.Close()
	}()

	ctx, cancel := context.WithCancel(ctx)
	s.cancelFunc = cancel

	for {
		err := s.processMessage(ctx, handler)
		if err != nil {
			switch {
			case errors.Is(ctx.Err(), context.Canceled):
				return ctx.Err()
			case errors.Is(err, ErrMsgIgnore):
			default:
				time.Sleep(errSleepTime)
			}
		}
	}
}

func (s *sub) processMessage(ctx context.Context, handler pubsub.MessageHandler) error {
	defer func() {
		if err := recover(); err != nil {
			debug.PrintStack()
		}
	}()

	msg, err := s.Receive(ctx)
	if err != nil {
		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
			return nil
		}
		return err
	}
	return handler(msg.Context(), msg)
}

func NewSubscriber(opt SubOptions) (pubsub.Subscriber, error) {
	if opt.Logger == nil {
		opt.Logger = log.DefaultNopLogger()
	}
	client, err := pulsar.NewClient(pulsar.ClientOptions{
		URL: opt.URL,
	})
	if err != nil {
		return nil, errors.New("failed to create pulsar client")
	}

	consumer, err := client.Subscribe(pulsar.ConsumerOptions{
		ReceiverQueueSize: opt.ReceiverQueueSize,
		Topic:             opt.Topic,
		SubscriptionName:  opt.SubscriptionName,
		Type:              opt.SubscriptionType,
	})
	if err != nil {
		return nil, errors.New("failed to create pulsar consumer")
	}

	return &sub{
		client:        client,
		consumer:      consumer,
		notPtMsg:      opt.NotPtMessage,
		ackType:       opt.AckType,
		subscribeName: opt.SubscriptionName,
		topic:         opt.Topic,
		spanName:      opt.Topic + opt.SubscriptionName,
	}, nil
}
